package com.wpen.unit.core.service.impl;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import com.wpen.unit.AppVariables;
import com.wpen.unit.core.cache.CacheDatas;
import com.wpen.unit.core.ddo.TriggerDo;
import com.wpen.unit.core.service.TriggerPoolHelper;
import com.wpen.unit.cron.CronUtil;
import com.wpen.unit.enums.DependType;
import com.wpen.unit.enums.ExecStatus;
import com.wpen.unit.enums.ScheduleType;
import com.wpen.unit.enums.YN;
import com.wpen.unit.infra.mybatis.job.po.BatchExecLogPo;
import com.wpen.unit.infra.mybatis.job.po.BatchInfoPo;
import com.wpen.unit.infra.mybatis.job.po.DependonPo;
import com.wpen.unit.infra.mybatis.job.po.JobExec1LogPo;
import com.wpen.unit.infra.mybatis.job.po.JobInfoPo;
import com.wpen.unit.infra.mybatis.job.po.JobUndoPlanPo;
import com.wpen.unit.infra.mybatis.job.po.TaskExecLogPo;
import com.wpen.unit.infra.mybatis.job.po.TaskInfoPo;
import com.wpen.unit.infra.mybatis.job.po.TaskPlanPo;
import com.wpen.unit.infra.repository.job.BatchExecLogRepository;
import com.wpen.unit.infra.repository.job.DependonRepository;
import com.wpen.unit.infra.repository.job.JobExec1LogRepository;
import com.wpen.unit.infra.repository.job.JobUndoPlanRepository;
import com.wpen.unit.infra.repository.job.TaskExecLogRepository;
import com.wpen.unit.infra.repository.job.TaskPlanRepository;
import com.wpen.unit.utils.IdUtil;
import com.wpen.unit.utils.JsonUtil;

import lombok.extern.slf4j.Slf4j;

/**
 * 
 * 
 * @author Wang Peng
 * @date 2022-12-06
 */
@Service
@Slf4j
public class TriggerPoolHelperImpl implements TriggerPoolHelper {

	@Autowired
	private CacheDatas cacheDatas;
	@Autowired
	private DependonRepository dependonRepository;
	@Autowired
	private BatchExecLogRepository batchExecLogRepository;
	@Autowired
	private TaskPlanRepository taskPlanRepository;
	@Autowired
	private TaskExecLogRepository taskExecLogRepository;
	@Autowired
	private JobExec1LogRepository jobExec1LogRepository;
	@Autowired
	private JobUndoPlanRepository jobUndoPlanRepository;

	@Override
	@Transactional
	public void triggerBatch(TriggerDo batchTriggerDo) {
		String acctDate = batchTriggerDo.getAcctDate();
		Long batchTriggerTime = batchTriggerDo.getTriggerNextTime();
		Long nowTime = batchTriggerDo.getNowTime();
		log.info("TriggerPool-triggerBatch received [" + batchTriggerDo.getDesc() + "=" + batchTriggerTime + "]");

		BatchInfoPo batchInfoPo = cacheDatas.getBatchInfoByCode(batchTriggerDo.getBizCode(),
				batchTriggerDo.getBatchCode());
		// 存入执行日志表，执行计划
		BatchExecLogPo batchExecLogPo = new BatchExecLogPo();
		long batchExecLogId = IdUtil.nextId();
		batchExecLogPo.setId(batchExecLogId);
		batchExecLogPo.setBizCode(batchInfoPo.getBizCode());
		batchExecLogPo.setBatchCode(batchInfoPo.getBatchCode());
		batchExecLogPo.setBatchTriggerTime(batchTriggerTime);
		batchExecLogPo.setBatchType(batchInfoPo.getBatchType());
		batchExecLogPo.setScheduleType(batchTriggerDo.getPlanType());
		batchExecLogPo.setExecDate(acctDate);
		batchExecLogPo.setStartupTime(new Date());
		batchExecLogPo.setExecStatus(ExecStatus.Running.value());
		batchExecLogPo.setExecNum(1);
		batchExecLogPo.setCreateTime(new Date());

		// 查询依赖
		List<DependonPo> dependonPos = dependonRepository.queryByParent(batchInfoPo.getBizCode(),
				batchInfoPo.getBatchCode(), null, null, DependType.Task);
		boolean hasDepend = false;
		if (dependonPos != null && !dependonPos.isEmpty()) {
			// 存在依赖，查询基础数据，触发
			for (DependonPo dependonPo : dependonPos) {
				boolean dpdTrigger = triggerDependon(batchExecLogId, dependonPo, batchInfoPo, batchTriggerTime, nowTime,
						acctDate);
				if (!hasDepend && dpdTrigger) {
					hasDepend = true;
				}
			}
		}
		if (!hasDepend) {
			// 没有依赖任务，，直接完成
			batchExecLogPo.setExecStatus(ExecStatus.Finished.value());
			batchExecLogPo.setEndTime(new Date());
			batchExecLogPo.setExecResult("No tasks to be performed.");
		}
		batchExecLogRepository.insert(batchExecLogPo);
	}

	@Override
	@Transactional
	public boolean triggerDependon(long batchLogId, DependonPo dependonPo, BatchInfoPo batchInfoPo,
			Long batchTriggerTime, Long nowTime, String acctDate) {
		TaskInfoPo taskInfoPo = cacheDatas.getTaskInfoByCode(dependonPo.getBizCode(), dependonPo.getBatchCode(),
				dependonPo.getTaskCode());
		if (taskInfoPo == null) {
			log.warn("Task[" + dependonPo.getBizCode() + "-" + dependonPo.getBatchCode() + "-"
					+ dependonPo.getTaskCode() + "] not found.");
			return false;
		}
		// 判断状态是否需要运行？？ 是否自动运行
		if (YN.No.value().equals(taskInfoPo.getEnabled())
				|| ScheduleType.Manual.value().equals(taskInfoPo.getScheduleType())) {
			return false;
		}
		// 判断一下 task的触发时间，确定是立刻执行还是延时执行。
		Long taskTriggerTime = batchTriggerTime;
		if (ScheduleType.Auto.value().equals(taskInfoPo.getScheduleType())) {
			taskTriggerTime = taskTriggerTime + 1000; // 延后一秒
		} else {
			// 定时执行
			String taskCron = taskInfoPo.getScheduleConf();
			Long batchMaxTime = batchTriggerTime
					+ (batchInfoPo.getExeTimeout() < 2 ? 2 : batchInfoPo.getExeTimeout()) * 60 * 1000;
			taskTriggerTime = CronUtil.getTimestamp(taskCron, new Date(batchTriggerTime));
			// 执行时间有效性判断： task有效时间 > 当前batch执行时间 && task有效时间 < batch最大执行时间
			if (taskTriggerTime < batchTriggerTime || taskTriggerTime > batchMaxTime) {
				// 时间配置的不对，暂不支持
			}
		}
		TaskPlanPo taskPlanPo = new TaskPlanPo();
		taskPlanPo.setId(IdUtil.nextId());
		taskPlanPo.setBatchExecLogId(batchLogId);
		taskPlanPo.setBizCode(taskInfoPo.getBizCode());
		taskPlanPo.setBatchCode(taskInfoPo.getBatchCode());
		taskPlanPo.setTaskCode(taskInfoPo.getTaskCode());
		taskPlanPo.setPlanType(taskInfoPo.getScheduleType());
		taskPlanPo.setExecDate(acctDate);
		taskPlanPo.setBatchTriggerTime(batchTriggerTime);
		taskPlanPo.setCreateTime(new Date());
		taskPlanPo.setValidStatus(YN.Yes.value());
		Date taskRunTime = null;
		if (taskTriggerTime < (batchTriggerTime + 2000)) {
			// 2秒内的立刻执行
			taskPlanPo.setTriggerNextTime(taskTriggerTime);
			taskRunTime = new Date(taskTriggerTime);
		} else {
			// 延时执行 ，放入触发队列
			Date nextTime = CronUtil.getTime(taskInfoPo.getScheduleConf(), new Date(batchTriggerTime));
			taskPlanPo.setTriggerNextTime(nextTime.getTime());
			taskRunTime = nextTime;
		}
		log.info("TriggerPool-triggerBatch save taskPlan[" + taskRunTime + "-" + batchTriggerTime + "]");
		taskPlanRepository.insert(taskPlanPo);
		return true;
	}

	@SuppressWarnings("unchecked")
	@Override
	@Transactional(propagation = Propagation.REQUIRES_NEW)
	public void triggerTask(TriggerDo taskTriggerDo) {
		String acctDate = taskTriggerDo.getAcctDate();
		Long batchTriggerTime = taskTriggerDo.getBatchTriggerTime();
		Long nowTime = taskTriggerDo.getNowTime();

		TaskPlanPo taskPlanPo = taskPlanRepository.queryListByCodes(taskTriggerDo.getBizCode(),
				taskTriggerDo.getBatchCode(), taskTriggerDo.getTaskCode(), taskTriggerDo.getBatchExecLogId(),
				taskTriggerDo.getBatchTriggerTime());
		if (taskPlanPo == null) {
			log.error("not found task paln[" + taskTriggerDo.getBizCode() + "-" + taskTriggerDo.getBatchCode() + "-"
					+ taskTriggerDo.getTaskCode() + "]");
		}
		if (YN.No.value().equals(taskPlanPo.getValidStatus())) {
			return;
		}

		log.info("TriggerPool-triggerTask received a new task[" + taskTriggerDo.getDesc() + "=" + nowTime + "]");

		// 判断：执行计划的类型，都是一次性的，执行完了关闭
		TaskPlanPo upTaskPlanPo = new TaskPlanPo();
		upTaskPlanPo.setId(taskPlanPo.getId());
		upTaskPlanPo.setValidStatus(YN.No.value());
		// 更新数据库，下一次执行时间
		taskPlanRepository.updateById(upTaskPlanPo);

		TaskInfoPo taskInfoPo = cacheDatas.getTaskInfoByCode(taskPlanPo.getBizCode(), taskPlanPo.getBatchCode(),
				taskPlanPo.getTaskCode());

		// 对象，任务日志表
		TaskExecLogPo taskLogPo = new TaskExecLogPo();
		taskLogPo.setBatchExecLogId(taskPlanPo.getBatchExecLogId());
		taskLogPo.setBizCode(taskInfoPo.getBizCode());
		taskLogPo.setBatchCode(taskInfoPo.getBatchCode());
		taskLogPo.setTaskCode(taskInfoPo.getTaskCode());
		taskLogPo.setBatchTriggerTime(batchTriggerTime);
		taskLogPo.setTaskTriggerTime(nowTime);
		taskLogPo.setScheduleType(taskInfoPo.getScheduleType());
		taskLogPo.setExecDate(acctDate);
		taskLogPo.setStartupTime(new Date());
		taskLogPo.setExecStatus(ExecStatus.Running.value());
		taskLogPo.setExecNum(1);
		taskLogPo.setCreateTime(new Date());

		// 查询依赖
		List<DependonPo> dependonPos = dependonRepository.queryByParent(taskInfoPo.getBizCode(),
				taskInfoPo.getBatchCode(), taskInfoPo.getTaskCode(), null, DependType.Job);
		if (dependonPos == null || dependonPos.isEmpty()) {
			taskLogPo.setExecStatus(ExecStatus.Finished.name());
			taskLogPo.setExecResult("No jobs need to be executed");
			taskExecLogRepository.insert(taskLogPo);
			return;
		}
		List<DependonPo> taskOwns = new ArrayList<>();
		// 只处理组内的，不支持跨组
		for (DependonPo dependonPo : dependonPos) {
			if (dependonPo.getBizCode().equals(taskInfoPo.getBizCode())
					&& dependonPo.getBatchCode().equals(taskInfoPo.getBatchCode())
					&& dependonPo.getTaskCode().equals(taskInfoPo.getTaskCode())) {
				taskOwns.add(dependonPo);
			}
		}
		// 每个机器数量
		Map<String, Integer> mapAzVms = new HashMap<>();
		mapAzVms.put(AppVariables.AZ1, taskInfoPo.getAz1InstNums());

		Map<String, Integer> mapAzDbCounts = new HashMap<>();
		Map<String, List<TaskExecLogPo>> mapAzTasks = new HashMap<>();
		// ADD 任务日志表记录
		List<TaskExecLogPo> listTaskExecLogPos = new ArrayList<>();
		taskLogPo.setId(IdUtil.nextId());
		listTaskExecLogPos.add(taskLogPo);
		mapAzDbCounts.put(AppVariables.AZ1, 1);
		mapAzTasks.put(AppVariables.AZ1, listTaskExecLogPos);
		// ADD 任务作业日志表记录
		List<JobExec1LogPo> listJobExec1LogPos = new ArrayList<>();
		// 亲生的 ，，生成任务作业记录
		for (DependonPo dependonPo : taskOwns) {
			// 执行机器---任务--并行线程-作业分片关系
			// 查询任务作业信息表，查询参数
			JobInfoPo jobInfoPo = cacheDatas.getJobInfoByCode(dependonPo.getBizCode(), dependonPo.getBatchCode(),
					dependonPo.getTaskCode(), dependonPo.getJobCode());

			Map<String, Object> mapParams = new HashMap<>();
			if (jobInfoPo.getNormParams() != null && !"".equals(jobInfoPo.getNormParams())) {
				mapParams = (Map<String, Object>) JsonUtil.toObject(jobInfoPo.getNormParams(), Map.class);
			}
			// 分析一下，是不是并行任务
			if (jobInfoPo.getNormShardingParams() != null && !"".equals(jobInfoPo.getNormShardingParams())
					&& jobInfoPo.getNormShardingParams().split(",").length > 1) {
				// 是可以进行并发执行的
				List<JobExec1LogPo> listOneTaskJobs = compNormalJobShardingParams(jobInfoPo, mapAzDbCounts, mapAzVms,
						mapParams, jobInfoPo.getNormShardingParams(), mapAzTasks, acctDate,
						taskPlanPo.getBatchExecLogId());
				// 加入到待入库列表
				listJobExec1LogPos.addAll(listOneTaskJobs);
			} else {
				// 单机器执行任务
				mapParams.put(AppVariables.SHARDING_KEY, jobInfoPo.getNormShardingParams());
				String jobParam = JsonUtil.toJson(mapParams);
				for (TaskExecLogPo taskExecLogPo : listTaskExecLogPos) {
					JobExec1LogPo jobExec1LogPo = new JobExec1LogPo();
					jobExec1LogPo.setId(IdUtil.nextId());
					jobExec1LogPo.setBatchExecLogId(taskPlanPo.getBatchExecLogId());
					jobExec1LogPo.setTaskExecLogId(taskExecLogPo.getId());
					jobExec1LogPo.setBizCode(jobInfoPo.getBizCode());
					jobExec1LogPo.setBatchCode(jobInfoPo.getBatchCode());
					jobExec1LogPo.setTaskCode(jobInfoPo.getTaskCode());
					jobExec1LogPo.setJobCode(jobInfoPo.getJobCode());
					jobExec1LogPo.setBatchTriggerTime(batchTriggerTime);
					jobExec1LogPo.setTaskTriggerTime(nowTime);
					jobExec1LogPo.setExecDate(acctDate);
					jobExec1LogPo.setExecParams(jobParam);
					jobExec1LogPo.setStartupTime(new Date());
					jobExec1LogPo.setExecStatus(ExecStatus.Init.value());
					jobExec1LogPo.setExecNum(1);
					jobExec1LogPo.setCreateTime(new Date());
					// 加入到待入库列表
					listJobExec1LogPos.add(jobExec1LogPo);
				}
			}
		}
		/******************************************/

		// ADD 任务作业undo表
		List<JobUndoPlanPo> listJobUndoPlans = new ArrayList<>();
		for (int i = 0; i < listJobExec1LogPos.size(); i++) {
			JobExec1LogPo jobExec1LogPo = listJobExec1LogPos.get(i);
			JobUndoPlanPo jobUndoPlanPo = new JobUndoPlanPo();
			jobUndoPlanPo.setId(jobExec1LogPo.getId());
			jobUndoPlanPo.setAppMode(taskInfoPo.getAppMode());
			jobUndoPlanPo.setAppJarName(taskInfoPo.getAppJarName());
			jobUndoPlanPo.setJobTriggerTime(nowTime + i);
			jobUndoPlanPo.setExecStatus(ExecStatus.Init.value());
			jobUndoPlanPo.setCreateTime(new Date());
			listJobUndoPlans.add(jobUndoPlanPo);
		}
		// 保存所有数据
		taskExecLogRepository.insertBatch(listTaskExecLogPos);
		jobExec1LogRepository.insertBatch(listJobExec1LogPos);
		jobUndoPlanRepository.insertBatch(listJobUndoPlans);
	}

	private List<JobExec1LogPo> compNormalJobShardingParams(JobInfoPo jobInfoPo, Map<String, Integer> mapAzDbCounts,
			Map<String, Integer> mapAzVms, Map<String, Object> mapParams, String shardingKey,
			Map<String, List<TaskExecLogPo>> mapAzTasks, String acctDate, long batchExecLogId) {
		List<JobExec1LogPo> listJobExec1LogPos = new ArrayList<>();
		// Assert.notNull(shardingKey, "这个肯定不是null, 并且split后是多个");
		String[] shardKeys = shardingKey.split(",");
		// 开始计算参数个数与 总机器并行数。。。。
		String azCode = AppVariables.AZ1;
		Integer azVmNum = mapAzVms.get(azCode);
		List<TaskExecLogPo> oneOfAzTaskTaskLogs = mapAzTasks.get(azCode);

		int dist = azVmNum / 1;
		if (azVmNum == 0) {
			mapParams.put(AppVariables.SHARDING_KEY, shardingKey);
			String jobParam = JsonUtil.toJson(mapParams);
			if (jobParam != null) {
				jobParam = jobParam.replaceAll("\n", "").replaceAll("\t", "");
			}
			// 直接失败，az下所有都失败
			for (TaskExecLogPo taskExecLogPo : oneOfAzTaskTaskLogs) {
				JobExec1LogPo jobExec1LogPo = new JobExec1LogPo();
				jobExec1LogPo.setId(IdUtil.nextId());
				jobExec1LogPo.setBatchExecLogId(batchExecLogId);
				jobExec1LogPo.setTaskExecLogId(taskExecLogPo.getId());
				jobExec1LogPo.setBizCode(jobInfoPo.getBizCode());
				jobExec1LogPo.setBatchCode(jobInfoPo.getBatchCode());
				jobExec1LogPo.setTaskCode(jobInfoPo.getTaskCode());
				jobExec1LogPo.setJobCode(jobInfoPo.getJobCode());
				jobExec1LogPo.setExecDate(acctDate);
				jobExec1LogPo.setBatchTriggerTime(taskExecLogPo.getBatchTriggerTime());
				jobExec1LogPo.setTaskTriggerTime(taskExecLogPo.getTaskTriggerTime());
				jobExec1LogPo.setExecParams(jobParam);
				jobExec1LogPo.setExecStatus(ExecStatus.Excption.value());
				jobExec1LogPo.setExecNum(1);
				jobExec1LogPo.setStartupTime(new Date());
				jobExec1LogPo.setCreateTime(new Date());
				jobExec1LogPo.setExecResult("No virtual machine available in the az[" + azCode + "]");
				listJobExec1LogPos.add(jobExec1LogPo);
			}
		} else {
			// 有可用机器大于等于2个，，shardKeys大于等于2个
			// int inte = shardKeys.length / dist;
			// int part = shardKeys.length % dist;
			Map<Integer, String> mapSks = new HashMap<>();
			if (dist <= 1) {
				mapSks.put(1, shardingKey);
			} else {
				for (int i = 0; i < shardKeys.length; i++) {
					int pointer = i % dist;
					String sk = mapSks.get(pointer);
					if (sk == null) {
						sk = shardKeys[i];
					} else {
						sk = sk + "," + shardKeys[i];
					}
					mapSks.put(pointer, sk);
				}
			}
			// 拆分成并行任务
			for (TaskExecLogPo taskExecLogPo : oneOfAzTaskTaskLogs) {
				for (Map.Entry<Integer, String> subParams : mapSks.entrySet()) {
					String subParam = subParams.getValue();
					mapParams.put(AppVariables.SHARDING_KEY, subParam);
					String jobParam = JsonUtil.toJson(mapParams);
					if (jobParam != null) {
						jobParam = jobParam.replaceAll("\r\n", "").replaceAll("\n", "").replaceAll("\t", "");
					}

					JobExec1LogPo jobExec1LogPo = new JobExec1LogPo();
					jobExec1LogPo.setId(IdUtil.nextId());
					jobExec1LogPo.setBatchExecLogId(batchExecLogId);
					jobExec1LogPo.setTaskExecLogId(taskExecLogPo.getId());
					jobExec1LogPo.setBizCode(jobInfoPo.getBizCode());
					jobExec1LogPo.setBatchCode(jobInfoPo.getBatchCode());
					jobExec1LogPo.setTaskCode(jobInfoPo.getTaskCode());
					jobExec1LogPo.setJobCode(jobInfoPo.getJobCode());
					jobExec1LogPo.setExecDate(acctDate);
					jobExec1LogPo.setBatchTriggerTime(taskExecLogPo.getBatchTriggerTime());
					jobExec1LogPo.setTaskTriggerTime(taskExecLogPo.getTaskTriggerTime());
					jobExec1LogPo.setExecParams(jobParam);
					jobExec1LogPo.setExecStatus(ExecStatus.Init.value());
					jobExec1LogPo.setExecNum(1);
					jobExec1LogPo.setStartupTime(new Date());
					jobExec1LogPo.setCreateTime(new Date());
					listJobExec1LogPos.add(jobExec1LogPo);
				}
			}
		}

		return listJobExec1LogPos;
	}

}
